package com.lxw.multidb.modules.sql;

import com.lxw.multidb.controller.dto.DbConnectionRequest;
import io.micrometer.common.util.StringUtils;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.List;

/**
 * Mysql2Kingbase
 *
 * @author lxw
 * @version V1.0
 * @date 2025/7/28 15:11
 */
@Service
@RequiredArgsConstructor
@Slf4j
public class Mysql2Kingbase {

    private final DatabaseMetaService databaseMetaService;

    public void toData(String tableName) {
        DbConnectionRequest sourceDb = new DbConnectionRequest();
        sourceDb.setDbType("mysql");
        sourceDb.setHost("192.168.1.134");
        sourceDb.setPort(3306);
        sourceDb.setUsername("root");
        sourceDb.setPassword("Sunway!963info");
        sourceDb.setDatabaseName("db_sxzz");
        sourceDb.setTableName(tableName);


        DbConnectionRequest tgtDb = new DbConnectionRequest();
        tgtDb.setDbType("kingbase");
        tgtDb.setHost("113.207.101.75");
        tgtDb.setPort(54321);
        tgtDb.setUsername("root");
        tgtDb.setPassword("Sunwayinfo.com.cn");
        tgtDb.setDatabaseName("db_sxzz");
        tgtDb.setTableName(tableName);

        // 从sourceDb 中，循环读取数据，并插入到tgtDb中

        //来源表名称
        String sourceTableName = sourceDb.getTableName();
        String condition = sourceDb.getCondition();

        // 查一字段列表
        List<String> columns = databaseMetaService.listColumns(sourceDb);
        if (columns.isEmpty()) {
            log.error("表{}无字段，终止。", tableName);
            return;
        }
        columns.remove("id");
        // ---------- Step 1 查询总记录数 ----------
        String countSql = "SELECT COUNT(1) FROM `" + sourceTableName + "` ";
        if (StringUtils.isNotBlank(condition)) {
            countSql += " where " + condition;
        }

        DataSource dataSource = databaseMetaService.createDataSource(sourceDb);
        DataSource tgtDataSource = databaseMetaService.createDataSource(tgtDb);
        // 根据参数查询数据量。如果数量小于等于，则停止
        long totalCount = 0L;
        try (Connection conn = dataSource.getConnection();
             Statement stmt = conn.createStatement();
             ResultSet rs = stmt.executeQuery(countSql)) {
            rs.next();
            totalCount = rs.getLong(1);
        } catch (Exception e) {
            log.error("===>数据库查询待收割总数失败！{}", e.getMessage(), e);
        }
        if (totalCount <= 0) {
            log.info("本次无数据需要收割。");
            return;
        }
        log.info("===>数据库收割，总记录数：{}", totalCount);


        // 构造插入SQL
        StringBuilder insertSql = new StringBuilder();
        insertSql.append("INSERT INTO ").append(tableName).append(" (");
        insertSql.append(String.join(",", columns));
        insertSql.append(") VALUES (");
        insertSql.append("?,".repeat(columns.size()));
        insertSql.setLength(insertSql.length() - 1); // 去掉最后一个逗号
        insertSql.append(")");


        StringBuilder sql = new StringBuilder();
        sql.append("select * ");
        sql.append(" from `").append(sourceTableName).append("` ");
        sql.append(" where 1=1 ");

        if (StringUtils.isNotBlank(condition)) {
            sql.append(" and ").append(condition);
        }
        long surplusCount = totalCount;
        //分批处理数据，并加入批次明细表
        int pageSize = 1000;
        int totalPage = (int) Math.ceil(totalCount / (double) pageSize);
        String lastId = "0";
        for (int page = 0; page < totalPage; page++) {
            // ---------- Step 2 构造分页 SQL ----------
            String querySql = sql + " and id > " + lastId;
            querySql += " ORDER BY id limit 0," + pageSize;
            log.info("===>分页查询执行SQL: {}", querySql);

            //  查询数据,并组装BizUploadInfo
            try (Connection conn = dataSource.getConnection();
                 PreparedStatement ps = conn.prepareStatement(querySql);
                 ResultSet rs = ps.executeQuery();
                 Connection tgtConn = tgtDataSource.getConnection();
                 PreparedStatement tgtPs = tgtConn.prepareStatement(insertSql.toString())) {

                tgtConn.setAutoCommit(false);

                int batchCount = 0;
                while (rs.next()) {
                    Object value = rs.getObject("id");
                    lastId = value.toString();
                    for (int i = 0; i < columns.size(); i++) {
                        Object value2 = rs.getObject(columns.get(i));
                        if (value2 instanceof String) {
                            // 过滤掉所有非法的0x00字符
                            value2 = ((String) value2).replaceAll("\u0000", "");
                        }
                        tgtPs.setObject(i + 1, value2);
                    }
                    tgtPs.addBatch();
                    batchCount++;
                    if (batchCount % 500 == 0) {
                        tgtPs.executeBatch();
                        tgtConn.commit();
                    }
                }
                if (batchCount % 500 != 0) {
                    tgtPs.executeBatch();
                    tgtConn.commit();
                }
                surplusCount -= pageSize;
                log.info("第{}批次迁移完成，记录数：{}，剩余数：{}", page + 1, batchCount, surplusCount);
            } catch (Exception e) {
                log.error("第{}批次迁移失败: {}", page + 1, e.getMessage(), e);
            }
        }

        log.info("数据迁移完成，总记录数：{}", totalCount);
    }


}
